package gatperdut.chato.core.messaging;

import gatperdut.chato.core.ChatoServer;

import java.io.IOException;

import org.apache.log4j.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.rabbitmq.client.ShutdownSignalException;

public class RPCServer extends Thread {

    private final Logger logger = Logger.getLogger( RPCServer.class );

    private final ChatoServer chatoServer;

    private static final String RPC_QUEUE_NAME = "rpc_queue";

    private Connection connection;

    private Channel channel = null;

    private QueueingConsumer consumer;

    private MessageInterpreter messageInterpreter;

    public RPCServer( ChatoServer chatoServer ) {
        this.chatoServer = chatoServer;
    }

    private void setup() throws IOException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost( "localhost" );
        setConnection( factory.newConnection() );
        setChannel( getConnection().createChannel() );
        getChannel().queueDeclare( RPC_QUEUE_NAME, true, false, false, null );
        getChannel().basicQos( 1 );
        setConsumer( new QueueingConsumer( getChannel() ) );
        getChannel().basicConsume( RPC_QUEUE_NAME, false, getConsumer() );
        setMessageInterpreter( new MessageInterpreter( getChatoServer() ) );
    }

    @Override
    public void interrupt() {
        getLogger().info( " RPC Server - Stopped." );
        if ( getConnection() != null ) {
            try {
                getConnection().close();
            }
            catch ( Exception ignore ) {
            }
        }
        super.interrupt();
    }

    @Override
    public void run() { // TODO does this need synchronization?
        try {
            setup();
            getLogger().info( " RPC Server - Awaiting RPC requests." );
            while ( true ) {
                Delivery delivery = getConsumer().nextDelivery();
                getLogger().info( " Something!" );
                BasicProperties props = delivery.getProperties();
                BasicProperties replyProps = new BasicProperties.Builder().correlationId( props.getCorrelationId() ).build();
                String message = new String( delivery.getBody(), "UTF-8" );
                String response = getMessageInterpreter().exec( message );
                if ( response == null ) { // TODO what is Nack doing exactly?
                    getChannel().basicNack( delivery.getEnvelope().getDeliveryTag(), false, false );
                }
                else {
                    getChannel().basicPublish( "", props.getReplyTo(), replyProps, response.getBytes( "UTF-8" ) );
                    getChannel().basicAck( delivery.getEnvelope().getDeliveryTag(), false );
                }
            }
        }
        catch ( IOException | ShutdownSignalException | ConsumerCancelledException | InterruptedException e ) {
            e.printStackTrace();
        }
    }

    public Logger getLogger() {
        return logger;
    }

    public ChatoServer getChatoServer() {
        return chatoServer;
    }

    public static String getRpcQueueName() {
        return RPC_QUEUE_NAME;
    }

    public Connection getConnection() {
        return connection;
    }

    public Channel getChannel() {
        return channel;
    }

    public QueueingConsumer getConsumer() {
        return consumer;
    }

    public MessageInterpreter getMessageInterpreter() {
        return messageInterpreter;
    }

    public void setConnection( Connection connection ) {
        this.connection = connection;
    }

    public void setChannel( Channel channel ) {
        this.channel = channel;
    }

    public void setConsumer( QueueingConsumer consumer ) {
        this.consumer = consumer;
    }

    public void setMessageInterpreter( MessageInterpreter messageInterpreter ) {
        this.messageInterpreter = messageInterpreter;
    }
}